package com.doit.demo.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @DATE 2022/2/17/14:27
 * @Author MDK
 * @Version 2021.2.2
 **/
public class IterateDemo {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        DataStreamSource<String> line = env.socketTextStream("linux01", 8888);

        //将数字转成long类型
        SingleOutputStreamOperator<Long> numbers = line.map(Long::parseLong);

        //调用iterate()方法 DataStream->IterativeStream 对numbers进行迭代
        IterativeStream<Long> iteration = numbers.iterate();

        //IterativeStream -> DataStream
        //对迭代出来的数据进行运算  //对输入的数据应用更新模型,输入数据的处理逻辑
        SingleOutputStreamOperator<Long> map = iteration.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                System.out.println("iterate input => " + value);
                return value -= 3;
            }
        });

    }
}
